package storm;

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
 * 
 * @author 李岩飞
 * @email eliyanfei@126.com
 * @date 2018年6月8日上午11:09:22
 */
public class LogIPBolt extends BaseRichBolt {
	private static final long serialVersionUID = 1L;
	OutputCollector _collector;

	public LogIPBolt() {
	}

	public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
		_collector = collector;
		ipMap = new HashMap<String, AtomicInteger>();
	}

	Map<String, AtomicInteger> ipMap = null;

	public void execute(Tuple tuple) {
		LogBean logBean = (LogBean) tuple.getValue(0);
		AtomicInteger v = ipMap.get(logBean.ip);
		if (v == null) {
			v = new AtomicInteger(0);
			ipMap.put(logBean.ip, v);
		}
		v.addAndGet(1);
		_collector.emit(new Values(logBean));
		_collector.ack(tuple);
	}
	
	@Override
	public void cleanup() {
		super.cleanup();
		for (Entry<String, AtomicInteger> entry : ipMap.entrySet()) {
			System.out.println("\tIP=" + entry.getKey() + ", count=" + entry.getValue().get());
		}
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("ip1"));
	}
}